package com.flume.ModifyFilter;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;

public class AddHeaderFilter implements Interceptor{
public static Logger logger = LoggerFactory.getLogger(AddBodyTimeFilter.class);
	
	private String field;
	private String encoding;
	private String newFiled;
	
	public AddHeaderFilter(String field, String encoding, String newField) {
		this.field = field;
		this.encoding = encoding;
		this.newFiled = newField;
	}
	
	public void initialize() {
		// TODO Auto-generated method stub
		logger.info("origin field :{}, add field :{}.",field,newFiled);
	}

	public Event intercept(Event event) {
		// TODO Auto-generated method stub
		if(field==null||"".equals(field)||newFiled==null||"".equals(newFiled)) {
			logger.warn("field or newField not set");
			return event;
		}
		
		try {
			JSONObject eventBody = JSONObject.parseObject(new String(event.getBody(), encoding));
			
			event.getHeaders().put(newFiled, eventBody.getString(field));
			
		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			logger.error("unsupported encoding : {}",encoding);
			e.printStackTrace();			
		} catch (Exception e) {
			logger.error("error: {}",e);
			e.printStackTrace();
		}
		
		
		return event;
	}

	public List<Event> intercept(List<Event> events) {
		// TODO Auto-generated method stub
		List<Event> result = new ArrayList<Event>();
		for(Event e:events) {
			result.add(intercept(e));
		}
		return result;
	}

	public void close() {
		// TODO Auto-generated method stub
		
	}

	/**
	 * 
	 * @author lenovo
	 * @describe FIELD  原始消息中需要截取的时间字段，如为空，则返回原始消息
	 * 			TIMETYPE 原始消息中时间字段类型，默认为plaintext，目前支持两种格式：日期字符串（2020-10-21 10:30:30）、unixtime(1603098465488)
	 * 			FORMAT 转换后的日期格式，默认为yyyy-MM-dd
	 * 			ENCODING 字符编码，默认为utf8
	 * 			NEWFIELD 新增字段名称，如为空，则返回原始消息
	 */
	public static class Builder implements Interceptor.Builder{

		public static String FIELD="field";
		public static String ENCODING="encoding";
		public static String NEWFIELD = "newField";
		
		private String field;
		private String encoding;
		private String newField;
		
		public void configure(Context context) {
			// TODO Auto-generated method stub
			field = context.getString(FIELD, null);
			encoding = context.getString(ENCODING, "UTF-8");
			newField = context.getString(NEWFIELD,field);
		}

		public Interceptor build() {
			// TODO Auto-generated method stub
			return new AddHeaderFilter(field, encoding, newField);
		}
		
	}
}
